#include <iostream>
#include <csignal>
#include <cstring>
#include <chrono>
#include <string>
#include <ctime>
#include <list>
#include <condition_variable>
#include <unordered_map>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <sys/time.h>
#include <pthread.h>
#include <sys/un.h>
#include <thread>
#include <log.h>
#include <mutex>
#include <fcntl.h>
#include <unistd.h>
#include <sys/file.h>
#include <mosquitto.h>

#include "tzsptap2pcap.hpp"
#include "common.hpp"
#include "arg.h"
#include "ev.h"
#include "config.h"

using namespace std;
zlog_category_t *z_cate = nullptr; /* zlog 描述符 */
typedef struct recv_data_struct {
public:
	recv_data_struct()
	{
		gettimeofday(&current_time, nullptr);
		recvlen = 0;
	}
	recv_data_struct(char *ip_in, int port_in, ssize_t len_in = 0)
	{
		gettimeofday(&current_time, nullptr);
		ipaddr = string(ip_in);
		port = port_in;
		recvlen = len_in;
		if (recvlen > 0) {
			recvbuf = new uint8_t[recvlen];
		}
	}
	~recv_data_struct()
	{
		if (recvlen > 0 && recvbuf) {
			delete[] recvbuf;
		}
	}
	string ipaddr;
	int port;
	timeval current_time;
	uint8_t *recvbuf;
	ssize_t recvlen;
	int pkg_type;
} recv_data_t;

typedef struct socket_struct_ctx {
public:
        ~socket_struct_ctx() {
                ev_io_stop(loop, &io);
                if (fd > 0) {
                        close(fd);
                        fd = -1;
                }
        }
	struct ev_loop *loop;
	struct ev_io io;
	int fd;
	string ipaddr;
	int port;
	int socket_type;
	int pkg_type;
	tzsptap2pcap *device;

} socket_ctx_t;

typedef struct socket_access_ctrl_ctx {
	struct ev_loop *loop;
	struct ev_io io;
	struct ev_timer _timer;
	int fd;
	int socket_type;
	string ipaddr;
	int port;
	int pkg_type;
} socket_ac_ctx_t;

unordered_map< string, tzsptap2pcap*> tzsptap2pcap_devices; //抓包保存的对象

uint32_t tzsptap2pcap_devices_count = 0;

list<socket_ctx_t *> socket_clients;
uint32_t client_count;

struct ev_loop *loop = nullptr;

bool MQTT_cmd_start = false;//MQTT 发来的指令

void tzsptap2pcap_devices_destroy()
{
	if (tzsptap2pcap_devices.empty()) {
		return;
	}
	sys_debug("Clean tzsptap2pcap_devices(size:%lu).\n", tzsptap2pcap_devices.size());
	for (auto& pair : tzsptap2pcap_devices) {
		if (pair.second) {
			delete pair.second;
			pair.second = nullptr;
		}
	}
	tzsptap2pcap_devices.clear();
}

list<recv_data_t *> recv_data_list;
mutex recv_data_list_mtx;
condition_variable recv_data_list_cv;

void recv_data_list_destroy()
{
	if (recv_data_list.empty()) {
		return;
	}
	list<recv_data_t *>::iterator it = recv_data_list.begin();
	while (it != recv_data_list.end()) {
		delete *it;
		it = recv_data_list.erase(it);
	}
}

void push_back_to_recv_data_list(recv_data_t *_input)
{
	unique_lock<mutex> lock(recv_data_list_mtx);
	recv_data_list.push_back(_input);
	recv_data_list_cv.notify_one();
}

/* unordered_map 格式的TZSP设备集合,格式为tzsptap2pcap_devices["dev_ip"]= class tzsptap2pcap;
   1、检查是否存在此ip地址的TZSP设备，如果没有，则需要新建一个，然后添加到tzsptap2pcap_devices中,键值中的键为IP，值为tzsptap2pcap的对象
   2、将接收到的数据(TZSP格式)插入到对应IP的处理设备的待处理list中
   */
tzsptap2pcap* new_add_tzsptap2pcap_device(string &ipaddr)
{
        tzsptap2pcap *new_device = new tzsptap2pcap(ipaddr, args_info.pcap_file_size_arg);//创建tzsptap2pcap处理对象

        tzsptap2pcap_devices[ipaddr] = new_device;
        tzsptap2pcap_devices_count++;
        return new_device;
}
tzsptap2pcap* new_add_tzsptap2pcap_device(const char *ipaddr)
{
	string ipaddr_string(ipaddr);
	return new_add_tzsptap2pcap_device(ipaddr_string);
}

tzsptap2pcap* find_tzsptap2pcap_device(string &ipaddr)
{
	auto it = tzsptap2pcap_devices.find(ipaddr);
	if (it == tzsptap2pcap_devices.end()) {
		return nullptr;
	} else {
		return tzsptap2pcap_devices[ipaddr];
	}
}
tzsptap2pcap* find_tzsptap2pcap_device(const char *ipaddr)
{
	string ipaddr_string = string(ipaddr);
	return find_tzsptap2pcap_device(ipaddr_string);
}

void find_tzsptap2pcap_device_push_data(recv_data_t *recv_data)
{
	tzsptap2pcap *tzsptap2pcap_dev = find_tzsptap2pcap_device(recv_data->ipaddr);
        if (tzsptap2pcap_dev == nullptr) {
		sys_debug("Not found ip:%s in tzsptap2pcap_devices, new and add it \n", recv_data->ipaddr.c_str());
                tzsptap2pcap_dev = new_add_tzsptap2pcap_device(recv_data->ipaddr);
	        tzsptap2pcap_dev->pkg_flag = recv_data->pkg_type;
                tzsptap2pcap_dev->set_pcap_save_top_dir(args_info.pcap_path_arg);//设置TOP dir
	        tzsptap2pcap_dev->start_thread(); //开启线程
        }
	tzsptap2pcap_dev->srcpkg_list_push_back(recv_data->recvbuf, recv_data->recvlen, recv_data->current_time);
}

atomic<bool> stopFlag(false);
void thread_pull_recv_data_list()
{
	while (true) {
		unique_lock<mutex> lock(recv_data_list_mtx);
		// 等待任务队列非空或者停止标志置位
		recv_data_list_cv.wait(lock, [] { return!recv_data_list.empty() || stopFlag;});
		if (!recv_data_list.empty()) {
			recv_data_t *recv_data = recv_data_list.front();
			recv_data_list.pop_front();
			find_tzsptap2pcap_device_push_data(recv_data);
			delete recv_data;
		} else if (stopFlag) {
			break;
		}
	}
	return;
}

void ev_UDP_recv_cb(EV_P_ struct ev_io *w, int revents)
{
	socket_ac_ctx_t *ctx = container_of(w, socket_ac_ctx_t, io);
	if (!ctx) {
		return;
	}

	int fd = ctx->fd;
	struct sockaddr_in from_addr;
	socklen_t from_len = sizeof(from_addr);
	memset(&from_addr, 0, sizeof(from_addr));
	uint8_t recvbuf[UDP_RECV_MAX] = {0};
	ssize_t recvlen;
	recvlen = recvfrom(fd, recvbuf, sizeof(recvbuf), 0, (struct sockaddr*)&from_addr, &from_len);
	if (recvlen <= 0) {
                sys_err("recvfrom error:%d(%s)!\n", errno, strerror(errno));
		return;
	}
        if (!MQTT_cmd_start) {
		if (!args_info.auto_start_arg) {
			/* 未开启抓包开关，直接退出 */
			return;
		}
        }
	char frome_ip[17] = {0};
	inet_ntop(AF_INET, &from_addr.sin_addr, frome_ip, 17);
	sys_debug("recvfrom UDP (IP:%s) %lu Bytes\n", frome_ip, recvlen);
	Lzlog_debug("recvfrom UDP (IP:%s) %lu Bytes", frome_ip, recvlen);
	Lhzlog_debug(recvbuf, recvlen);
	//dump_hex (recvbuf, recvlen, 16);
	printf("\n");
	recv_data_t *recv_data = new recv_data_t(frome_ip, ntohs(from_addr.sin_port), recvlen);
	recv_data->pkg_type = ctx->pkg_type;
	if (recv_data) {
		if (recv_data->recvbuf) {
			memcpy(recv_data->recvbuf, recvbuf, recvlen);
			push_back_to_recv_data_list(recv_data);
		} else {
			delete recv_data;
		}
	}
}

void ev_TCP_recv_cb(EV_P_ struct ev_io *w, int revents)
{
	socket_ctx_t *clictx = container_of(w, socket_ctx_t, io);
	if (!clictx) {
		return;
	}
	int fd = clictx->fd;

	uint8_t recvbuf[TCP_RECV_MAX] = {0};
	ssize_t recvlen;
	recvlen = recv(fd, recvbuf, TCP_RECV_MAX, 0);
	if (!recvlen) {
		sys_debug("client %s is disconnected, close it!\n", clictx->ipaddr.c_str());
		/* 删除 客户端client
		 * 处理未保存的文件并停止处理
		 * 删除对应tzsptap2pcap处理单元
		 * */
		socket_clients.remove(clictx);
		delete clictx;
		return;
	} else if (recvlen < 0) {
		if (errno != EAGAIN && errno != EWOULDBLOCK) {
			sys_debug("recv failed! %d(%s), close it!\n", errno, strerror(errno));
			socket_clients.remove(clictx);
			delete clictx;
		}
		return;
	}  /* End if */
        if (!MQTT_cmd_start) {
                /* 未开启抓包开关，直接退出 */
                return;
        }

	sys_debug("recvfrom TCP client (IP:%s) %lu Bytes\n", clictx->ipaddr.c_str(), recvlen);
	dump_hex (recvbuf, recvlen, 16);
	timeval current_time;
	gettimeofday(&current_time, nullptr);
	if (clictx->device) {
		clictx->device->srcpkg_list_push_back(recvbuf, recvlen, current_time);
	}
}

void ev_accept_cb(EV_P_ struct ev_io *w, int revents)
{
	socket_ac_ctx_t *ctx = container_of(w, socket_ac_ctx_t, io);
	if (!ctx) {
		return;
	}
	int fd = ctx->fd;

	struct sockaddr_in cli_addr;
	socklen_t len = sizeof(cli_addr);
	int cli_fd = accept(fd, (struct sockaddr*)&cli_addr, &len);
	if (cli_fd == -1) {
		sys_err("accept failed:%s(%d)\n", strerror(errno), errno);
		return;
	}
        if (!MQTT_cmd_start) {
                close(cli_fd);
                /* 未开启抓包开关，直接退出 */
                return;
        }
        net_optimize_enableQuickAck(cli_fd); //设置快速ACK
	char client_ip[17] = {0};
	inet_ntop(AF_INET, &cli_addr.sin_addr, client_ip, 17);

	sys_debug("Accept new client connect IP:%s, port:%d\n", client_ip , cli_addr.sin_port);

	tzsptap2pcap *tzsptap2pcap_dev = find_tzsptap2pcap_device((const char *)client_ip);
        if (tzsptap2pcap_dev == nullptr) {
                tzsptap2pcap_dev = new_add_tzsptap2pcap_device((const char *)client_ip);//创建tzsptap2pcap处理对象
                tzsptap2pcap_dev->pkg_flag = ctx->pkg_type;
                tzsptap2pcap_dev->set_pcap_save_top_dir(args_info.pcap_path_arg);//设置TOP dir
                tzsptap2pcap_dev->start_thread(); //开启线程
        }
	setnonblocking(cli_fd); //设置非阻塞
        net_optimize_setReceiveBuffer(cli_fd, 16 * MB); //设置TCP接收缓冲区 16MB

	socket_ctx_t *clictx = new socket_ctx_t;
	clictx->loop = EV_A;
	clictx->fd = cli_fd;
	clictx->ipaddr = string(client_ip);
	clictx->device = tzsptap2pcap_dev;

	ev_io_init(&clictx->io, ev_TCP_recv_cb, cli_fd, EV_READ);
	ev_io_start(EV_A_ &clictx->io);
	socket_clients.push_back(clictx);
}

void ev_socket_ac_timer_cb(EV_P_ struct ev_timer *w, int revents)
{
	socket_ac_ctx_t *ctx = container_of(w, socket_ac_ctx_t, _timer);
	if (!ctx) {
		return;
	}
	if (ctx->fd != -1) {
		return;
	}
	int fd = create_bind_net_socket(ctx->socket_type, ctx->ipaddr.c_str(), ctx->port);
	if (fd == -1) {
		ev_timer_again(EV_A_ w);
		return;
	}
	ctx->fd = fd;
	if (ctx->socket_type == SOCK_STREAM) {
		sys_debug("Use TCP to recv data.\n");
		ev_io_init(&ctx->io, ev_accept_cb, fd, EV_READ);
	} else if (ctx->socket_type == SOCK_DGRAM) {
		sys_debug("Use UDP to recv data,fd:%d.\n", ctx->fd);
		ev_io_init(&ctx->io, ev_UDP_recv_cb, fd, EV_READ);
	}
	ev_io_start(EV_A_ &ctx->io);
}

/* ====================== MQTT========================= */
struct mosquitto *mosq = NULL;
int isconnect = 0;
int get_mosq_connect_state()
{
	return isconnect;
}
/* MQTT log cb */
void my_log_callback(struct mosquitto *mosq, void *obj, int level, const char *str)
{
	sys_debug("%s\n", str);
}
/* MQTT 连接cb */
void my_connect_callback(struct mosquitto *mosq, void *obj, int result)
{
	isconnect = 1;
	sys_debug("connect callback: %d\n", result);
	if (args_info.subtopic_given) {
		mosquitto_subscribe(mosq, NULL, args_info.subtopic_arg, args_info.subqos_arg);
	}
}
/* MQTT 断开cb （触发重联）*/
void my_disconnect_callback(struct mosquitto *mosq, void *obj, int rc)
{
	isconnect = 0;
	rc = mosquitto_reconnect(mosq);
	sys_debug("mosquitto_reconnect: %d\n", rc);
}
/* MQTT 发送数据cb */
void my_publish_callback(struct mosquitto *mosq, void *obj, int mid)
{
	sys_debug("publish callback: %d\n", mid);
}
/* MQTT 发送数据 */
void publish_msg(struct mosquitto *_mosq, void *data, int len)
{
	mosquitto_publish(_mosq, NULL, args_info.topic_arg, len, data, args_info.qos_arg, 0);
}
/* 解析来自MQTT的指令 */
void recv_pkg_control_cmd_parse(const char *cmd_str, uint16_t cmd_len)
{
        Json::Value JsonRoot;
        Json::Value root;
        Json::CharReaderBuilder Rbuilder;
        Rbuilder["collectComments"] = false;  // 不收集注释
        Rbuilder["allowComments"] = true;     // 允许JSON中存在注释
        Rbuilder["strictRoot"] = false;       // 不强制根为对象或数组
        Rbuilder["allowDroppedNullPlaceholders"] = true;  // 允许省略的null值
        Rbuilder["allowNumericKeys"] = true;  // 允许数字作为对象键
        string errors;
        unique_ptr<Json::CharReader> reader(Rbuilder.newCharReader());

        string op_string;
        bool parse_flag = reader->parse( cmd_str, cmd_str + cmd_len, &JsonRoot, &errors);
        // 处理解析结果
        if (!parse_flag) {
                sys_err("JSON cmd from MQTT 解析失败:%s\n", errors.c_str());
                root["result"] = "-1";
                root["desc"] = "Parse cmd string to JSON failed!";
                goto parse_return;
        }
        if (!JsonRoot.isMember("op")) {
                root["result"] = "-1";
                root["desc"] = "Done not get key [op] in cmd!";
                goto parse_return;
        }
        op_string = JsonRoot["op"].asString();
        root["op"] = op_string;
        if (op_string == "start") {
                if (MQTT_cmd_start == true) {
                        root["result"] = "-1";
                        root["desc"] = "Start failed, another package capture is performed!";
                } else {
                        MQTT_cmd_start = true;
                        root["result"] = "0";
                        root["desc"] = "Start to recv TZSP and shark pkg and save file!";
                }

        } else if (op_string == "stop") {
                if (MQTT_cmd_start == false && !args_info.auto_start_arg) {
                        root["result"] = "-1";
                        root["desc"] = "Stop failed, Didn't start!";
                } else {
                        MQTT_cmd_start = false;
                        root["device_count"] = to_string(tzsptap2pcap_devices.size());
                        Json::Value devarray(Json::arrayValue);
                        for (const auto& pair : tzsptap2pcap_devices) {
                                const std::string& key = pair.first; // 设备名称
                                tzsptap2pcap* device = pair.second; // 设备指针

                                if (device) {
                                        devarray.append(device->pcap_files_formatJSON());
                                }
                        }
                        root["devices"] = devarray;

                        for (const auto& pair : tzsptap2pcap_devices) {
                                delete pair.second; // 设备指针
                        }

                        tzsptap2pcap_devices.clear();

                        if (socket_clients.size()) {
                                for (auto it = socket_clients.begin(); it != socket_clients.end(); ) {
                                        delete *it;
                                        it = socket_clients.erase(it);
                                }
                        }
                        socket_clients.clear();
                        root["result"] = "0";
                        root["desc"] = "Stop recv PKG!";
                }
        } else if (op_string == "query" || op_string == "state") {
                if (MQTT_cmd_start) {
                        root["state"] = "1";
                        root["desc"] = "Started!";
                        root["device_count"] = to_string(tzsptap2pcap_devices.size());
                        Json::Value devarray(Json::arrayValue);
                        for (const auto& pair : tzsptap2pcap_devices) {
                                const std::string& key = pair.first; // 设备名称
                                tzsptap2pcap* device = pair.second; // 设备指针
                                if (device) {
                                        devarray.append(device->pcap_files_formatJSON());
                                }
                        }
                        root["devices"] = devarray;
                } else {
                        root["state"] = "2";
                        root["desc"] = "Not Start!";
                }

        } else {
                root["result"] = "-1";
                root["desc"] = "Unknow op!";
                goto parse_return;
        }

parse_return:
        // 将JSON对象转换为字符串
        Json::StreamWriterBuilder Wbuilder;
        Wbuilder["commentStyle"] = "None";//关闭注释
        Wbuilder["indentation"] = "  ";  // 缩进2 字符
        Wbuilder["emitUTF8"] = true;// 使用UTF-8编码

        string jsonString = Json::writeString(Wbuilder, root);
        publish_msg(mosq, (void *)jsonString.c_str(), jsonString.size());
        return;

}
/* MQTT 接收msg cb */
#define CMDLEN	(2048)
void my_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message)
{
	if(message->payloadlen <= 0) {
		return;
	}
	sys_debug("%s %s\n", (char *)message->topic, (char *)message->payload);
	Lzlog_debug("Get message from MQTT:%s", (char *)message->payload);
        recv_pkg_control_cmd_parse((const char *)message->payload, strlen((char *)message->payload));

}
/* MQTT 设置用户密码 */
int my_pw_callback(char *buf, int size, int rwflag, void *userdata)
{
	if(!args_info.keypassword_given)
		return 0;

	int passlen = strlen(args_info.keypassword_arg);
	if(size < passlen) {
		sys_err("key password too long: %d %d\n", passlen, size);
		exit(-1);
	}
	memcpy(buf, args_info.keypassword_arg, passlen);
	return passlen;
}
/*  信号处理 */
void My_signal_handler(int signum)
{
	sys_debug("Get signal signum:%d, Clean and exit!\n", signum);
	Lzlog_debug("Get signal signum:%d, Clean and exit!", signum);
	stopFlag = true;
	recv_data_list_cv.notify_all();

	tzsptap2pcap_devices_destroy();
	recv_data_list_destroy();
	while (!socket_clients.empty()) {
		delete socket_clients.front();
		socket_clients.pop_front();
	}
	if (EV_A) {
		ev_loop_destroy(EV_A);
	}
        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        mosq = nullptr;
        exit(signum);
}

static int one_instance()
{
	char pidfile[256];
	snprintf(pidfile, 256, "/var/run/tzsptap2pcap.pid");
	int pid_file = open(pidfile, O_CREAT | O_RDWR | O_CLOEXEC, 0666);
	int rc = flock(pid_file, LOCK_EX | LOCK_NB);
	if(rc) {
		if(EWOULDBLOCK == errno) {
			sys_warn("Another mqttdiris running\n");
			exit(1);
		}
	}
	return 0;
}
/* 优化设备TCP 连接性能 */
static void sysctrl_net_optimize()
{
        system("sysctl -w net.core.rmem_default=16777216"); // 统级默认接收缓冲区大小
        system("sysctl -w net.core.rmem_max=67108864");     // 系统级最大接收缓冲区大小
        system("sysctl -w net.ipv4.tcp_rmem='4096 16777216 67108864'");// TCP 协议专用的接收缓冲区控制

        system("sysctl -w net.ipv4.tcp_window_scaling=1");  // 启用窗口缩放优化
        system("sysctl -w net.ipv4.tcp_moderate_rcvbuf=1"); // 启用接收缓冲区自动调优
        system("sysctl -w net.ipv4.tcp_sack=1");            // 启用选择性ACK

        system("sysctl -w net.core.netdev_max_backlog=30000"); // 增大网络设备接收队列
        system("sysctl -w net.ipv4.tcp_low_latency=1");        // 低延迟模式
}

int main (int argc, char *argv[])
{
	signal(SIGINT, My_signal_handler);
	signal(SIGTERM, My_signal_handler);
	proc_args(argc, argv);
	if (args_info.log_conf_given) {//初始化日志
		int rc = zlog_init(args_info.log_conf_arg);
		if (0 == rc) {
			z_cate = zlog_get_category("my_log");
		} else {
			sys_err("zlog init error!%s\n", strerror(errno));
			z_cate = nullptr;
		}
	} else {
		sys_debug("Not set zlog config!\n");
	}
	create_directory(args_info.pcap_path_arg);
	one_instance();
	if(daemon_mode) {
		daemon(1, 0);
	}
/*==================================================================*/
        int ret = -1;

	mosquitto_lib_init();
	mosq = mosquitto_new(args_info.id_arg, args_info.clean_session_flag, NULL);
	if(!mosq) {
		switch(errno) {
		case ENOMEM:
			sys_err("Error: Out of memory.\n");
			Lzlog_error("mosquitto_new error!Out of memory.");
			break;
		case EINVAL:
			sys_err("Error: Invalid id.\n");
			Lzlog_error("mosquitto_new error!nvalid id.");
			break;
		}
		exit(-1);
	}

	if (args_info.willtopic_given) {
		int payloadlen = 0;
		if (args_info.willpayload_given) {
			payloadlen = strlen(args_info.willpayload_arg);
		}
		ret = mosquitto_will_set(mosq, args_info.willtopic_arg,
			payloadlen, args_info.willpayload_arg,
			args_info.willqos_arg,
			args_info.willretain_arg);
		if (ret != MOSQ_ERR_SUCCESS) {
			sys_err("Error: mosquitto set will failed\n");
			Lzlog_error("mosquitto set will failed.");
			exit(-1);
		}
	}

	mosquitto_log_callback_set(mosq, my_log_callback);
	mosquitto_connect_callback_set(mosq, my_connect_callback);
	mosquitto_disconnect_callback_set(mosq, my_disconnect_callback);
	mosquitto_publish_callback_set(mosq, my_publish_callback);
	mosquitto_message_callback_set(mosq, my_message_callback);

	if (args_info.username_given && args_info.password_given) {
		sys_debug("set username: %s password: %s\n", args_info.username_arg, args_info.password_arg);
		Lzlog_debug("set username: %s password: %s", args_info.username_arg, args_info.password_arg);
		if (mosquitto_username_pw_set(mosq, args_info.username_arg, args_info.password_arg) != MOSQ_ERR_SUCCESS) {
			sys_err("set username password failed\n");
			Lzlog_error("mosquitto set username password failed.");
			exit(-1);
		}
	}

	if (args_info.cafile_given && args_info.cert_given && args_info.key_given) {
		mosquitto_tls_set(mosq, args_info.cafile_arg, NULL, args_info.cert_arg, args_info.key_arg, my_pw_callback);
	}

	mosquitto_tls_insecure_set(mosq, args_info.insecure_flag);
        sleep(1);
	ret = mosquitto_connect(mosq, args_info.host_arg, args_info.port_arg,
		args_info.keepalive_arg);
	if(ret != MOSQ_ERR_SUCCESS) {
		sys_err("mosquitto_connect failed: %s(%d)\n", strerror(errno), errno);
		Lzlog_error("mosquitto_connect failed: %s(%d)", strerror(errno), errno);
                mosquitto_destroy(mosq);
	        mosquitto_lib_cleanup();
                return 0;
	}

	sys_debug("mosquitto thread start\n");
	Lzlog_info("mosquitto thread start\n");
	ret = mosquitto_loop_start(mosq);
	if(ret != MOSQ_ERR_SUCCESS) {
		sys_err("mosquitto_loop_start failed\n");
		Lzlog_error("mosquitto_loop_start failed");
                mosquitto_destroy(mosq);
	        mosquitto_lib_cleanup();
                return 0;
	}
	Lzlog_info("Creat net packet capture and init it!");
/*==================================================================*/

	loop = EV_DEFAULT;
	//创建线程循环监听recv data list
	thread t_pull_recv_data_list(thread_pull_recv_data_list);
	t_pull_recv_data_list.detach();

	/* ====== TZSP AC ====== */
	socket_ac_ctx_t tzsp_ctx_ac;
	tzsp_ctx_ac.loop = EV_A;
	tzsp_ctx_ac.fd = -1;
	tzsp_ctx_ac.socket_type = SOCK_DGRAM;
	tzsp_ctx_ac.ipaddr = string(args_info.tzsp_bind_ip_arg);
	tzsp_ctx_ac.port = args_info.tzsp_bind_port_arg;
	tzsp_ctx_ac.pkg_type = PKG_TYPE_TZSP;

	ev_timer_init(&tzsp_ctx_ac._timer, ev_socket_ac_timer_cb, 1, 0);
	ev_timer_start(EV_A_ &tzsp_ctx_ac._timer);

	/* ====== Shark AC ====== */
	socket_ac_ctx_t shark_ctx_ac;
	shark_ctx_ac.loop = EV_A;
	shark_ctx_ac.fd = -1;
	shark_ctx_ac.socket_type = args_info.shark_type_arg;
	shark_ctx_ac.ipaddr = string(args_info.shark_bind_ip_arg);
	shark_ctx_ac.port = args_info.shark_bind_port_arg;
	shark_ctx_ac.pkg_type = PKG_TYPE_SHARK;

	ev_timer_init(&shark_ctx_ac._timer, ev_socket_ac_timer_cb, 1, 0);
	ev_timer_start(EV_A_ &shark_ctx_ac._timer);

	ev_run(EV_A_ 0);

	close(tzsp_ctx_ac.fd);
	close(shark_ctx_ac.fd);
	ev_loop_destroy(EV_A);

	while (!socket_clients.empty()) {
		delete socket_clients.front();
		socket_clients.pop_front();
	}
cleanup:
        mosquitto_destroy(mosq);
	mosquitto_lib_cleanup();

	return 1;
}
